#include "thread.h"

using namespace bhttp;

thread::thread(const std::string& name)
    : name_(name)
{
}

thread::~thread()
{
}

void thread::start()
{
    if (running_) return;
    running_ = true;
    thread_ = std::make_unique<std::thread>(std::bind(&thread::thread_func, this));
}

void thread::stop()
{
    if (!running_) return;
    {
        running_ = false;
        not_empty_.notify_all();
    }
    if (thread_ && thread_->joinable())
        thread_->join();
}

void thread::run(const task& tk)
{
    std::unique_lock<std::mutex> lk(mutex_);
    while (is_full())
        not_full_.wait(lk);
    queue_.emplace_back(tk);
    not_empty_.notify_one();
}

void thread::run(task&& tk)
{
    std::unique_lock<std::mutex> lk(mutex_);
    while (is_full())
        not_full_.wait(lk);
    queue_.emplace_back(std::move(tk));
    not_empty_.notify_one();
}

void thread::thread_func()
{
    try {
        while (running_)
        {
            //task tk = take();
            //if (tk) tk();
            while (task tk = take()) tk();
        }
    }
    catch (const std::exception& ex)
    {
        fprintf(stderr, "exception caught in bhttp::thread %s\n", name_.c_str());
        fprintf(stderr, "reason: %s\n", ex.what());
        abort();
    }
}

bool thread::is_full() const
{
    return max_queue_size_ > 0 && queue_.size() >= max_queue_size_;
}

thread::task thread::take()
{
    std::unique_lock<std::mutex> lk(mutex_);
    while (queue_.empty() && running_)
        not_empty_.wait(lk);
    task tk;
    if (!queue_.empty())
    {
        tk = queue_.front();
        queue_.pop_front();
        if (max_queue_size_ > 0)
            not_full_.notify_one();
    }
    return tk;
}
